package com.ys.servers.common.verticle;

import com.ys.servers.common.mq.EventMQ;
import com.ys.servers.common.mq.InternalEventData;
import io.vertx.circuitbreaker.CircuitBreaker;
import io.vertx.circuitbreaker.CircuitBreakerOptions;
import io.vertx.core.AbstractVerticle;
import io.vertx.core.CompositeFuture;
import io.vertx.core.Future;
import io.vertx.core.impl.ConcurrentHashSet;
import io.vertx.core.json.JsonObject;
import io.vertx.core.logging.Logger;
import io.vertx.core.logging.LoggerFactory;
import io.vertx.core.shareddata.AsyncMap;
import io.vertx.core.shareddata.SharedData;
import io.vertx.servicediscovery.Record;
import io.vertx.servicediscovery.ServiceDiscovery;
import io.vertx.servicediscovery.ServiceDiscoveryOptions;

import java.util.*;
import java.util.function.BiConsumer;

/**
 * Created by taotao on 2017/3/1.
 */
public abstract class BasicMicroServiceVerticle extends AbstractVerticle {

    protected static final Logger logger = LoggerFactory.getLogger(BasicMicroServiceVerticle.class);

    protected ServiceDiscovery discovery;
    protected CircuitBreaker circuitBreaker;
    protected Set<Record> registeredRecords = new ConcurrentHashSet<>();
    protected SharedData shareData;

    protected Map<String, BiConsumer<InternalEventData, Future<InternalEventData>>> internalHandlerMap = new HashMap<>();

    /**
     * event mq的发送地址
     *
     * @return
     */
    protected String address() {
        return getClass().getSimpleName() + "EventAddress";
    }

    /**
     * 服务发现中的名称
     *
     * @return
     */
    protected abstract String interName();

    @Override
    public void start() throws Exception {
        // init service discovery instance
        discovery = ServiceDiscovery.create(vertx, new ServiceDiscoveryOptions().setBackendConfiguration(config()));

        // init circuit breaker instance
        JsonObject cbOptions = config().getJsonObject("circuit-breaker") != null ?
                config().getJsonObject("circuit-breaker") : new JsonObject();
        circuitBreaker = CircuitBreaker.create(cbOptions.getString("name", "circuit-breaker"), vertx,
                new CircuitBreakerOptions()
                        .setMaxFailures(cbOptions.getInteger("max-failures", 5))
                        .setTimeout(cbOptions.getLong("timeout", 10000L))
                        .setFallbackOnFailure(true)
                        .setResetTimeout(cbOptions.getLong("reset-timeout", 30000L))
        );

        //  initStoreClient();
        // shareData = vertx.sharedData();
        if (interName() != null) {
            EventMQ eventMQ = EventMQ.getApiEventMQ(vertx);
            eventMQ.consumer(address(), (reqData, future) -> handlerInternalRequest(reqData, future));
            initInternalDispatcher();
            publishService(interName(), address());
        }

    }

    /**
     * 初始化服务间内部调用
     */
    protected abstract void initInternalDispatcher();

    /**
     * 发布服务
     *
     * @param name
     * @param address
     * @return
     */
    protected Future<Void> publishService(String name, String address) {
        Record record = new Record().setType("event-mq-server").setLocation(new JsonObject().put("address", address)).setName(name);
        return publish(record);
    }

    /**
     * Publish a service with record.
     *
     * @param record service record
     * @return async result
     */
    protected Future<Void> publish(Record record) {
        if (discovery == null) {
            try {
                start();
            } catch (Exception e) {
                throw new IllegalStateException("Cannot create discovery service");
            }
        }

        Future<Void> future = Future.future();

        discovery.publish(record, ar -> {
            if (ar.succeeded()) {
                registeredRecords.add(record);
                logger.info("Service <" + ar.result().getName() + "> published");
                future.complete();
            } else {
                future.fail(ar.cause());
            }
        });

        return future;
    }

    public void stop(Future<Void> future) throws Exception {
        List<Future> futures = new ArrayList<>();
        registeredRecords.stream().forEach(record -> {
            Future<Void> unregistrationFuture = Future.future();
            futures.add(unregistrationFuture);
            discovery.unpublish(record.getRegistration(), unregistrationFuture.completer());
        });

        if (futures.isEmpty()) {
            discovery.close();
            future.complete();
        } else {
            CompositeFuture.all(futures)
                    .setHandler(ar -> {
                        discovery.close();
                        if (ar.failed()) {
                            future.fail(ar.cause());
                        } else {
                            future.complete();
                        }
                    });
        }
    }

    /**
     * 发送服务之间的异步调用
     *
     * @param internalEventData
     * @return
     */
    protected Future<JsonObject> interService(InternalEventData internalEventData) {

        Future future = Future.future();
        discovery.getRecord(r -> r.getName().equals("inter-" + internalEventData.getRecordAddress()), ar -> {
            if (ar.succeeded()) {
                if (ar.result() != null) {
                    Record record = ar.result();
                    String address = record.getLocation().getString("address");
                    EventMQ eventMQ = EventMQ.getInterEventMQ(vertx);
                    eventMQ.send(address, internalEventData, respData -> future.complete(respData));
                } else {
                    System.out.print("inter service:no match service verticle address");
                    future.fail("inter service:no match service verticle address");
                }
            } else {
                System.out.print("lookup discovery record failed");
                future.fail(ar.cause());
            }
        });
        return future;
    }


    protected void handlerInternalRequest(InternalEventData internalEventData, Future<InternalEventData> respFuture) {
        BiConsumer<InternalEventData, Future<InternalEventData>> consumer = internalHandlerMap.get(internalEventData.getApiName());
        //logger.info("dispatch to function - msg id:" + reqSocketData.getMsgId());
        consumer.accept(internalEventData, respFuture);
    }


    protected Future<Object> getValue(String key, String mapName) {
        Future<Object> future = Future.future();
        if (vertx.isClustered()) {
            shareData.<String, Object>getClusterWideMap(mapName, res -> {
                if (res.succeeded()) {
                    AsyncMap<String, Object> map = res.result();
                    map.get(key, as -> {
                        if (as.succeeded()) {
                            future.complete(as.result());
                        } else {
                            logger.error(as.cause());
                            future.fail(as.cause());
                        }
                    });
                } else {
                    logger.error(res.cause());
                    future.fail(res.cause());
                }
            });
        } else {
            String userId = shareData.<String, String>getLocalMap(mapName).get(key);
            future.complete(userId);
        }

        return future;
    }

    protected Future<Void> setValue(String key, Object value, String mapName) {
        Future<Void> future = Future.future();
        if (vertx.isClustered()) {
            shareData.<String, Object>getClusterWideMap(mapName, res -> {
                if (res.succeeded()) {
                    AsyncMap<String, Object> map = res.result();
                    map.put(key, value, as -> {
                        if (as.succeeded()) {
                            System.out.println("set cluster map id to share data");
                            future.complete();
                        } else {
                            logger.error(as.cause());
                            future.fail(as.cause());
                        }
                    });
                } else {
                    logger.error(res.cause());
                    future.fail(res.cause());
                }
            });
        } else {
            shareData.<String, Object>getLocalMap(mapName).put(key, value);
            System.out.println("set local map id to share data");
            future.complete();
        }

        return future;
    }
}
